/**
 * Created by IntelliJ IDEA.
 * User: jiyufei
 * Date: 2018/10/19
 * Time: 15:37
 **/
package com.jiyufei.linux.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import com.jiyufei.linux.bean.MessageLog;
import com.jiyufei.linux.log.EsLogCenter;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class MessageSplitBolt extends BaseRichBolt {
    
    private OutputCollector outputCollector;
    private TopologyContext topologyContext;
    private String regStr = "(\\S+)\\s((\\d+\\.){3}\\d+)\\s(.+)\\s(\\d+)\\s(\\S+\\s+\\S+\\s\\S+)\\s(\\S+)\\s(\\S+?):\\s(.*)";
    private EsLogCenter esLogCenter;
    
    private Pattern pattern,pattern2;
    private Matcher matcher,matcher2;
    private SimpleDateFormat sdf1,sdf2,sdf3;
    
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputCollector = outputCollector;
        this.topologyContext = topologyContext;
        pattern = Pattern.compile(regStr);
        sdf1 = new SimpleDateFormat("yyyy");
        sdf2 = new SimpleDateFormat("yyyy MMM dd HH:mm:ss", Locale.ENGLISH);
        sdf3 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZZ");
        esLogCenter = new EsLogCenter();
    }
    
    @Override
    public void execute(Tuple tuple) {
        String log = tuple.getStringByField("log");
        if(log == null){
            esLogCenter.addLog("message",topologyContext.getTopologyId() + ":" + topologyContext.getThisComponentId(),"yunying",
                    "[HlasLinuxLog.MessageSplitBolt]","High","log is null." + log);
        }
        MessageLog messageLog = new MessageLog();
        matcher = pattern.matcher(log);
        if(matcher.find()){
            messageLog.setType(matcher.group(1));
            messageLog.setClientIp(matcher.group(2));
            messageLog.setPath(matcher.group(4));
            messageLog.setOffset(Integer.parseInt(matcher.group(5)));
            try {
                Date now = new Date();
                String year = sdf1.format(now);
                String time = year + " " + matcher.group(6);
                Date date1 = sdf2.parse(time);
                messageLog.setTime(sdf3.format(date1));
            }catch (Exception e){
                esLogCenter.addLog("message",topologyContext.getTopologyId() + ":" + topologyContext.getThisComponentId(),"yunying",
                        "[HlasLinuxLog.MessageSplitBolt.execute]","High","转换时间失败！" + log);
                messageLog.setTime(sdf3.format(new Date()));
            }
            messageLog.setHostname(matcher.group(7));
            messageLog.setProcess(matcher.group(8));
            messageLog.setMessage(matcher.group(9));
            messageLog.setLog(log);
            
            outputCollector.emit("message-stream",tuple,new Values(messageLog));
            
        }else{
            esLogCenter.addLog("message",topologyContext.getTopologyId() + ":" + topologyContext.getThisComponentId(),"yunying",
                    "[HlasLinuxLog.MessageSplitBolt.MessageSplitBolt]","High","正则匹配失败！" + log);
        }
        
        outputCollector.ack(tuple);
        
    }
    
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declareStream("message-stream",new Fields("message-log"));
    }
}
